package com.nuo.laboratory.juc.collection;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.junit.Before;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.LongStream;
import java.util.stream.Stream;

/**
 * 用于一个线程生产对象，一个线程消费对象
 * <p>
 * BlockingQueue 阻塞有界队列   一般用于一个线程负责生产，一个负责消费
 * BlockingDeque 阻塞有界双端队列
 * ArrayBlockingQueue 基于数组有界阻塞队列 FIFO(先进先出) 实现 BlockingQueue
 * LinkedBlockingQueue 基于链接节点有界阻塞队列 FIFO(先进先出) 实现 BlockingQueue
 * LinkedBlockingDeque 基于链接节点有界阻塞双端队列 FIFO(先进先出) 实现 BlockingQueue，BlockingDeque
 * ConcurrentLinkedQueue 基于链接节点的无界线程安全队列 FIFO(先进先出)
 * Fork/Join ForkJoinPoll 分治算法 并行处理，等待处理结果合并 JDK1.7
 * CompletableFuture JDK1.8
 *
 * @author LGC
 * @date 2022/3/8 10:06
 */
public class QueueTest {

    ExecutorService execute = null;

    @Before
    public void init() {
        execute = new ThreadPoolExecutor(8, 1024, 600, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1024), ThreadFactoryBuilder.create().setNamePrefix("Thread-nuo-").build());
    }

    @Test
    public void arrayBlockingQueue() throws InterruptedException {
        BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(1024);
        execute.execute(() -> {
            try {
                for (int i = 1; i <= 10; i++) {
                    System.out.println("put " + i);
                    blockingQueue.put(i);
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        execute.execute(() -> {
            try {
                for (int i = 1; i <= 10; i++) {
                    System.out.println("take " + blockingQueue.take());

                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        // 阻塞主线程
        Thread.sleep(10 * 200);
    }

    @Test
    public void LinkedBlockingDeque() throws InterruptedException {

        BlockingDeque<Integer> deque = new LinkedBlockingDeque<>(2);
        deque.putFirst(1);
        deque.putLast(2);

        System.out.println(deque.takeLast());
        System.out.println(deque.takeFirst());
    }

    @Test
    public void future() throws Exception {
        Future<?> future = execute.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                Long start = System.currentTimeMillis();
                while (true) {
                    Long current = System.currentTimeMillis();
                    if ((current - start) > 1000) {
                        System.out.println("111111111111");
                        return 1;
                    }
                }
            }
        });
        //  调用get()方法获取任务结果,如果任务没有执行完成则阻塞等待
        Integer o = (Integer) future.get();
        System.out.println("future " + o);

    }

    @Test
    public void forkAndJoin() {
        ForkJoinPool forkJoinPool = new ForkJoinPool(4);
        long[] numbers = LongStream.rangeClosed(0, 2000000).toArray();
        ForkJoinTask<Long> forkJoinTask = new ForkJoinSumCalculator(numbers);
        Long sum = forkJoinPool.invoke(forkJoinTask);
        System.out.println(sum);
    }

    @Test
    public void completableFuture() {
        ConcurrentLinkedQueue<Integer> concurrentLinkedQueue = new ConcurrentLinkedQueue<>();

        CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {
            for (int i = 1; i <= 10; i++) {
                try {
                    System.out.println(Thread.currentThread().getName() + " offer " + i);
                    concurrentLinkedQueue.offer(i);
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return null;
        }, execute);

        CompletableFuture<Void> completableFuture1 = CompletableFuture.supplyAsync(() -> {
            for (int i = 1; i <= 10; i++) {
                try {
                    System.out.println(Thread.currentThread().getName() + " poll " + concurrentLinkedQueue.poll());
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return null;
        }, execute);

        CompletableFuture<Void> completableFuture2 = CompletableFuture.allOf(completableFuture, completableFuture1);

    }


    @Test
    public void test() {
        // thenApply
        CompletableFuture<Integer> futureA = CompletableFuture.supplyAsync(() -> 1, execute);
        CompletableFuture<Integer> futureB = futureA.thenApplyAsync(s -> s + 2, execute);
        CompletableFuture<String> future3 = futureB.thenApply(Object::toString);
        System.out.println(future3.join());

        // thenCompose
        CompletableFuture<Integer> futureAA = CompletableFuture.supplyAsync(() -> 1, execute);
        CompletableFuture<Object> futureBB = futureAA.thenCompose(i -> {
            return CompletableFuture.supplyAsync(() -> {
                return i + 2 + "";
            });
        });
        System.out.println(futureBB.join());

        // thenCombine
        // 1.获取单价 2.获取折扣 3.获取折扣价
        CompletableFuture<Double> futurePrice = CompletableFuture.supplyAsync(() -> 100d, execute);
        CompletableFuture<Double> futureDiscount = CompletableFuture.supplyAsync(() -> 0.8, execute);
        CompletableFuture<Double> futureResult = futurePrice.thenCombine(futureDiscount, (price, discount) -> price * discount);
        System.out.println("获取折扣价:" + futureResult.join());

    }


    @Test
    public void allOfTest() throws Exception {
        List<Integer> result = new ArrayList<>();
        CompletableFuture<List<Integer>> future1 = CompletableFuture.supplyAsync(() -> {
            List<Integer> list = new ArrayList<>();
            for (int i = 1; i <= 100; i++) {
                list.add(i);
                try {
                    Thread.sleep(200);
                    System.out.println(Thread.currentThread().getName() + "  " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
            return list;
        }, execute);

        CompletableFuture<List<Integer>> future2 = CompletableFuture.supplyAsync(() -> {
            List<Integer> list = new ArrayList<>();
            for (int i = 101; i <= 200; i++) {
                list.add(i);
                try {
                    Thread.sleep(200);
                    System.out.println(Thread.currentThread().getName() + "  " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return list;
        }, execute);

        CompletableFuture.allOf(future1, future1).join();
        result.addAll(future1.get());
        result.addAll(future2.get());
        System.out.println(result);
    }

    @Test
    public void stream() throws Exception {
        List<User> result = new ArrayList<>();
        result.add(new User("xiaoming",19));
        result.add(new User("xiaohua",18));
        Stream<User> userStream = result.stream().filter(new Predicate<User>() {
            @Override
            public boolean test(User user) {
                return user.getAge() > 18;
            }
        });
        Stream<String> stringStream1 = result.stream().map(new Function<User, String>() {
            @Override
            public String apply(User user) {
                return user.getName().toUpperCase(Locale.ROOT);
            }
        });

        Stream<String> stringStream = result.stream().flatMap(new Function<User, Stream<String>>() {
            @Override
            public Stream<String> apply(User user) {
                List<String> strList = new ArrayList<>();
                strList.add(user.getName() + ":" + user.getAge());
                return strList.stream();
            }
        });
    }
}

class A {
    private volatile static A a;

    public static A getAInstant() {
        if (a == null) {
            synchronized (A.class) {
                if (a == null) {
                    a = new A();
                }
            }
        }
        return a;
    }

}

@Data
@NoArgsConstructor
@AllArgsConstructor
class User {
    private String name;
    private Integer age;

}
